package com.spyz.zk.project2.zkcase;

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {

    private final ZooKeeper zk;
    private String connectString="master:2181,slave1:2181,slave2:2181";
    private int sessionTimeout=2000;

    private CountDownLatch connectLatch = new CountDownLatch(1);
    private String waitPath;
    private CountDownLatch waitLatch=new CountDownLatch(1);
    private String currentMode;

    public DistributedLock() throws IOException, InterruptedException, KeeperException {

        // 获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // connectLatch 如果连接上就可以释放
                if (watchedEvent.getState()==Event.KeeperState.SyncConnected){
                    connectLatch.countDown();
                }


                if (watchedEvent.getType()==Event.EventType.NodeDeleted&&watchedEvent.getPath().equals(waitPath)){
                    waitLatch.countDown();
                }
            }
        });

        // 等待zk连接成功，否则阻塞程序继续运行
        connectLatch.await();

        // 判断根节点/locks是否存在
        if(zk.exists("/locks",false)==null){
            zk.create("/locks", "lock".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

    }

    // 对zk加锁
    public void zklock()  {
        // 创建临时的带序号节点
        try {
            currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 判断创建的节点是否是最小的节点，如果是获取锁，否则监听前一个节点
            List<String> children = zk.getChildren("/locks", false);
            // 如果childern只有一个，则直接获取锁，否则判断谁最小
           if (children.size()==1){
               return;
           }else {
               Collections.sort(children);

               // 获取节点名称 seq-0000000000
               String thisNode = currentMode.substring("/locks/".length());
               // 通过seq-0000000000 获取该节点在列表中的位置
               int thisIndex = children.indexOf(thisNode);

               //判断
               if (thisIndex==-1){
                   System.out.println("数据异常");
               }else if (thisIndex==0){
                   // 如果是最小的节点，则直接获取锁
                   return;
               }else {
                   // 需要监听前一个节点
                   waitPath = "/locks/"+children.get(thisIndex-1);
                   zk.getData(waitPath, true,null);
                   waitLatch.await();
                   return;
               }
           }


        } catch (KeeperException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }


    }

    // 对zk解锁
    public void unZkLock(){

        // 删除节点
        try {
            zk.delete(currentMode,-1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (KeeperException e) {
            throw new RuntimeException(e);
        }
    }
}
